package com.fiona.remoting.netty.codec;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fiona.common.extension.ExtensionLoader;
import com.fiona.core.message.Message;
import com.fiona.core.message.RpcRequest;
import com.fiona.core.message.RpcResponse;
import com.fiona.core.message.enums.SerializationTypeEnum;
import com.fiona.core.protocol.MessageProtocol;
import com.fiona.remoting.netty.Constants;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

/**
 * custom protocol decoder
 * <pre>
 *   0     1     2     3     4     6     7     28    29   30
 *   +-----+-----+-----+-----+-----+-----+-----+-----+----+
 *   |          full length  |    messageKey   |type|codec|
 *   +----------------------------------------------------+
 *   |                                                    |
 *   |                    body                            |
 *   |                                                    |
 *   +----------------------------------------------------+
 * 4B full length（消息长度）    24B messageKey（消息ID）
 * 1B type（消息类型） 1B codec（序列化类型）
 * body（object类型数据）
 * </pre>
 * <p>
 * {@link LengthFieldBasedFrameDecoder} is a length-based decoder , used to solve TCP unpacking and sticking problems.
 * </p>
 *
 * @author huacongnan
 * @createTime on 2021/3/2
 * @see <a href="https://www.cnblogs.com/crazymakercircle/p/10294745.html">LengthFieldBasedFrameDecoder 秒懂</a>
 */
public class MessageDecoder extends LengthFieldBasedFrameDecoder {
	private static Logger logger = LoggerFactory.getLogger(MessageDecoder.class);
    public MessageDecoder() {
        this(Constants.MAX_FRAME_LENGTH, 0, 4, 0, 0);
    }

    /**
     * @param maxFrameLength      Maximum frame length. It decide the maximum length of data that can be received.
     *                            If it exceeds, the data will be discarded.
     * @param lengthFieldOffset   Length field offset. The length field is the one that skips the specified length of byte.
     * @param lengthFieldLength   The number of bytes in the length field.
     * @param lengthAdjustment    The compensation value to add to the value of the length field
     * @param initialBytesToStrip Number of bytes skipped.
     *                            If you need to receive all of the header+body data, this value is 0
     *                            if you only want to receive the body data, then you need to skip the number of bytes consumed by the header.
     */
    public MessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,
                             int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decoded = super.decode(ctx, in);
        if (decoded instanceof ByteBuf) {
            ByteBuf frame = (ByteBuf) decoded;
            if (frame.readableBytes() >= Constants.HEAD_LENGTH) {
                try {
                    return decodeFrame(frame);
                } catch (Exception e) {
                	logger.error("Decode frame error!", e);
                    throw e;
                } finally {
                    frame.release();
                }
            }

        }
        return decoded;
    }


    private Object decodeFrame(ByteBuf in) {
        int fullLength = in.readInt();
        String messageKey = readMessageKey(in);
        byte messageType = in.readByte();
        byte codecType = in.readByte();
        Message message = new Message();
        message.setCodec(codecType);
        message.setMessageKey(messageKey);
        message.setType(messageType);
        if (messageType == Constants.HEARTBEAT_REQUEST_TYPE) {
            message.setData(Constants.PING);
            return message;
        }
        if (messageType == Constants.HEARTBEAT_RESPONSE_TYPE) {
            message.setData(Constants.PONG);
            return message;
        }
        int bodyLength = fullLength - Constants.HEAD_LENGTH;
        if (bodyLength > 0) {
            byte[] bs = new byte[bodyLength];
            in.readBytes(bs);
            // deserialize the object
            String codecName = SerializationTypeEnum.getName(message.getCodec());
            logger.info("codec name: [{}] ", codecName);
            MessageProtocol protocol = ExtensionLoader.getExtensionLoader(MessageProtocol.class)
                    .getExtension(codecName);
            if (messageType == Constants.REQUEST_TYPE) {
                RpcRequest tmpValue = protocol.deserialize(bs, RpcRequest.class);
                message.setData(tmpValue);
            } else {
                RpcResponse tmpValue = protocol.deserialize(bs, RpcResponse.class);
                message.setData(tmpValue);
            }
        }
        return message;

    }


    private String readMessageKey(ByteBuf in) {
        int len = Constants.MESSAGE_KEY_LENGTH;
        byte[] tmp = new byte[len];
        in.readBytes(tmp);
        return new String(tmp);
    }

}
